package com.nbcio.iot.protocol.mqtt;

import com.nbcio.iot.config.IotMqttProperties;
import io.moquette.broker.Server;
import io.moquette.BrokerConstants;
import io.moquette.broker.config.IConfig;
import io.moquette.broker.config.MemoryConfig;
import io.moquette.broker.security.IAuthenticator;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * 内嵌 MQTT Broker（Moquette）带健康日志
 */
@Slf4j
@Component("embeddedMqttBroker")
@Order(0)
@RequiredArgsConstructor
public class EmbeddedMqttBroker {

    private final IotMqttProperties properties;
    private Server server;

    @PostConstruct
    public void start() {
        if (!properties.getEmbedded().isEnabled()) {
            log.info("[MQTT-BROKER] Embedded disabled by config");
            return;
        }
        final String host = properties.getEmbedded().getHost();
        final int port = properties.getEmbedded().getPort();
        final boolean allowAnonymous = properties.getEmbedded().isAllowAnonymous();

        log.info("[MQTT-BROKER] Starting embedded broker ... host={}, port={}, allowAnonymous={}, persistence=false",
                host, port, allowAnonymous);

        try {
            // Ensure 'data' directory exists to avoid FileNotFoundException for data/.moquette_uuid
            File dataDir = new File("data");
            if (!dataDir.exists()) {
                boolean created = dataDir.mkdirs();
                if (!created) {
                    log.warn("[MQTT-BROKER] Failed to create data directory at {}", dataDir.getAbsolutePath());
                } else {
                    log.info("[MQTT-BROKER] Ensured data directory at {}", dataDir.getAbsolutePath());
                }
            } else {
                log.info("[MQTT-BROKER] Using existing data directory at {}", dataDir.getAbsolutePath());
            }

            Properties p = new Properties();
            p.setProperty(BrokerConstants.HOST_PROPERTY_NAME, host);
            p.setProperty(BrokerConstants.PORT_PROPERTY_NAME, String.valueOf(port));
            p.setProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, String.valueOf(allowAnonymous));
            // disable persistence to simplify first run
            p.setProperty(BrokerConstants.PERSISTENCE_ENABLED_PROPERTY_NAME, "false");

            IConfig config = new MemoryConfig(p);
            server = new Server();

            IAuthenticator authenticator = null;
            if (!allowAnonymous && properties.getEmbedded().getUsername() != null) {
                String expectUser = properties.getEmbedded().getUsername();
                String expectPass = properties.getEmbedded().getPassword() == null ? "" : properties.getEmbedded().getPassword();
                authenticator = (clientId, username, password) -> {
                    String pwd = password == null ? "" : new String(password, StandardCharsets.UTF_8);
                    return expectUser.equals(username) && expectPass.equals(pwd);
                };
            }

            server.startServer(config, null, null, authenticator, null);
            log.info("[MQTT-BROKER] Embedded broker started");

            String testHost = "0.0.0.0".equals(host) ? "127.0.0.1" : host;
            boolean healthy = checkTcpReachable(testHost, port, 1500);
            if (healthy) {
                log.info("[MQTT-BROKER] BROKER HEALTH: OK (tcp://{}:{})", testHost, port);
            } else {
                log.warn("[MQTT-BROKER] BROKER HEALTH: FAIL (cannot connect tcp://{}:{})", testHost, port);
            }
        } catch (Exception e) {
            log.error("[MQTT-BROKER] FAILED to start embedded broker (host={}, port={})", host, port, e);
        }
    }

    @PreDestroy
    public void stop() {
        try {
            if (server != null) {
                server.stopServer();
                log.info("[MQTT-BROKER] Embedded broker stopped");
            }
        } catch (Exception ignore) { }
    }

    private boolean checkTcpReachable(String host, int port, int timeoutMs) {
        try (Socket socket = new Socket()) {
            socket.connect(new InetSocketAddress(host, port), timeoutMs);
            return true;
        } catch (IOException e) {
            return false;
        }
    }
}